package sparkDStream


  import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
  import org.apache.kafka.common.serialization.StringDeserializer
  import org.apache.spark.SparkConf
  import org.apache.spark.streaming.kafka010.KafkaUtils
  import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
  import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  import org.apache.spark.streaming.{Seconds, StreamingContext}
  import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  import java.util

  object HellosparkDStream {


    def main(args: Array[String]): Unit = {

      /*  maven 依赖  之前 spark-core
          spark streaming   spark-streaming jar
          spark-streaming-kafka

         1) 创建 sparkconf 配置项 本地，应用程序名字
         2） 创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
         3)  设置 编码  日志级别 error
         4）  设置 kafka topic ，key value 反序列化， groupid ，offset 提交
         5） 订阅 kafka的 消费者  topic ssc
         6） 消费topic 当中的数据

       */
      //创建 sparkconf 配置项 本地，应用程序名字
      val conf = new SparkConf().setMaster("local[*]").setAppName("helloStream")
      //创建 sparkStreaming  加载conf （ 实时的数据流）  ssc
      val ssc = new StreamingContext(conf,Seconds(2))

      // error
      ssc.sparkContext.setLogLevel("error")

      // kafka 配置
      val kfkParams = Map[String,Object](
        "bootstrap.servers" ->"cheng:9092",
        "key.deserializer"  -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> "cheng",
        "enable.auto.commit" -> (false:java.lang.Boolean)
      )

      // topic
      val topicName = Array("stuInfo")
      //    val topicName = Array("test1")

      // kafkaUtils  导入一个依赖 spark-streaming-kafka
      val streamRdd = KafkaUtils.createDirectStream[String,String](
        ssc,
        PreferConsistent, // 位置策略
        Subscribe[String,String](topicName,kfkParams)  // 订阅topic的名字
      )
      val line = streamRdd.map(_.value())
      val res = line.flatMap(_.split(" ")).map((_,1)).reduceByKeyAndWindow(
        _+_,
        Seconds(4),
        Seconds(4))
      res.foreachRDD(
        x=>{
          println("---------数据结果------------")
          x.foreach(println)
        }
      )
      //streamRdd  kafka 返回的对象，值只是在value    RDD（key,value ）

    /*  val line =streamRdd.foreachRDD(
        x=>{
          if(x.isEmpty()){
            val rs1 = x.map(_.value())
          }
        }
      )

     */
      // 窗口函数
      line
      //  producer 配置项
      val property = new util.HashMap[String,Object]()
      property.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.235.128:9092")
      property.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
      property.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
      streamRdd.foreachRDD(
        x=>{
          // 不为空的时候 进行打印
          if(!x.isEmpty()){
            // 取 kafka当中的 value ，为实际的数据
            val line = x.map(_.value())
            // 对一行数据进行打印
            //          line.foreach(println)
            // hello java
            // 词频统计
            val result = line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
            result.foreach(println)
            result.foreach(
              obj =>{
                // obj  元素

                val kfkProducer = new KafkaProducer[String,String](property)
                kfkProducer.send(   new ProducerRecord[String,String]    (  "kafka",obj.toString()   ))
                kfkProducer.close()


              }
            )
            // 窗口函数


          }
        }
      )
      ssc.start()  //开启 ssc
      ssc.awaitTermination() //等待数据输入


    }
}
